package com.zhangdi.mgsync.core.thread;

import com.mongodb.client.MongoChangeStreamCursor;
import com.mongodb.client.MongoCollection;
import com.mongodb.client.model.changestream.ChangeStreamDocument;
import com.mongodb.client.model.changestream.FullDocument;
import com.mongodb.client.model.changestream.OperationType;
import com.zhangdi.mgsync.core.cache.Cache;
import com.zhangdi.mgsync.core.common.MgSyncConst;
import com.zhangdi.mgsync.core.enums.SyncState;
import com.zhangdi.mgsync.core.rabbit.AdminTemplate;
import org.bson.BsonDocument;
import org.bson.BsonInvalidOperationException;
import org.bson.BsonTimestamp;
import org.bson.BsonType;
import org.bson.BsonValue;
import org.bson.Document;
import org.bson.types.ObjectId;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.amqp.core.QueueInformation;

import java.text.DateFormat;
import java.text.SimpleDateFormat;
import java.util.ArrayList;
import java.util.Date;
import java.util.HashMap;
import java.util.Iterator;
import java.util.List;
import java.util.Map;

/**
 * Created by zhangdi on 2020/07/16 20:36:13.
 */
public class WatchThread implements SyncWorker {
    Logger logger = LoggerFactory.getLogger(WatchThread.class);

    AdminTemplate adminTemplate;
    Cache<String, String> cache;

    volatile WorkerState workerState;


    private MongoChangeStreamCursor<ChangeStreamDocument<Document>> cursor = null;
    private final MongoCollection<Document> collection;
    private String collName;
    private int second;
    private DateFormat df = new SimpleDateFormat("yyyy-MM-dd hh:mm:ss");

    private final int unconsume_message_max_count = 750;
    private int allow_unconsume_message_count = 100;

    private int batchSize = 100;
    private long batch_timeout = 10 * 1000;
    private volatile boolean monitoring = true;
    private volatile boolean shutdown = true;
    private volatile long duration = 0;
    private volatile SyncState SendState = SyncState.idle;
    private volatile List<Map<String, Object>> dataBatchList = new ArrayList<>();
    private volatile List<Map<String, Object>> dataBatchBufferList = new ArrayList<>();


    int getSecond() {
        if (second == 0) {
            return (int) (new Date().getTime() / 1000L);
        }
        return second;
    }

    public WatchThread(AdminTemplate adminTemplate, MongoCollection<Document> collection, Cache<String, String> cache, int second) {
        this.adminTemplate = adminTemplate;
        this.collection = collection;
        this.second = second;
        this.cache = cache;
        this.collName = collection.getNamespace().getCollectionName();
        this.workerState = new WorkerState();
    }

    @Override
    public void run() {
        if (isRunning()) {
            logger.info("{} is running", collName);
            return;
        }

        try {
//            String token = cache.get(MgSyncConst.KEY + collName);
//            if (StringUtils.isBlank(token)) {
            //启动定时任务，30秒关闭一次，30秒后检查队列消息，如果没了就启动
            startAtTime();


//            } else {
//                start(token);
//            }

        } catch (Exception e) {
            workerState.setState(SyncState.pause);
            if(monitoring) {
//                WatchLooperFuture.cancel(true);
                adminTemplate.getEmailUtil().send(collName + "监听线程中断", "token=" + adminTemplate.getResumeToken(collName) +"  "+ e.getMessage());
                //这里出错是在idle超时后的，因此不能作为续传时间
                logger.error("{} Thread Interrupted", collName, e);
            } else {
                logger.info("{} 关闭监听", collName);
            }
        }
    }

    private boolean resumeTokenIsEmpty() {
        BsonDocument token = adminTemplate.getResumeToken(collName);
        return token == null || token.isEmpty();
    }

    //有必要这么多地方关闭连接吗
    public void stopListener() {
        QueueInformation queueInfo = adminTemplate.getQueueInfo(collName);
        if(queueInfo.getMessageCount() >= unconsume_message_max_count) {
            monitoring = false;
            logger.info("{} 暂停遍历..., count={}", collName, queueInfo.getMessageCount());
        }

        if(!monitoring && dataBatchList.size() == 0 && dataBatchBufferList.size() ==0) {
            if(cursor != null) {
                cursorClose();
            }
        }
    }

    public boolean tryReStartListener() {
        QueueInformation queueInfo = adminTemplate.getQueueInfo(collName);

        logger.info("{} 尝试重新监听,当前消息数量={}", collName, queueInfo.getMessageCount());

        return queueInfo.getMessageCount() <= allow_unconsume_message_count && !monitoring;
    }

    public void startAtTime() {
        logger.info("准备重新连接, cursor是null={}", cursor==null);
        shutdown = false;
        monitoring = true;
        workerState.setState(SyncState.running);

        try {
            if (resumeTokenIsEmpty()) {
                BsonTimestamp bsonTimestamp = new BsonTimestamp(getSecond(), 1);
                cursor = collection.watch(MgSyncConst.PIPELINE).
                        fullDocument(FullDocument.UPDATE_LOOKUP).
                        startAtOperationTime(bsonTimestamp).cursor();
                logger.info("{} Resume At Time={}", collName, getSecond());
            } else {
                cursor = collection.watch(MgSyncConst.PIPELINE).
                        fullDocument(FullDocument.UPDATE_LOOKUP).
                        startAfter(adminTemplate.getResumeToken(collName)).cursor();
                logger.info("{} Resume At Token={}", adminTemplate.getResumeToken(collName), getSecond());
            }
        } catch (Exception e) {
            if (cursor != null) {
                cursor.close();
            }
            logger.error("{} Start At Time {}, No Data.", collName, getSecond());
//            workerState.setState(SyncState.fail);
            second = 0;
            this.startAtTime();
            return;
        }

        EachCursor(cursor);
    }

    private void EachCursor(MongoChangeStreamCursor<ChangeStreamDocument<Document>> cursor) {
        if (cursor == null) {
            logger.error("{} Cursor is null.", collName);
            return;
        }

        while (monitoring && cursor.hasNext()) {
            if (!isRunning()) {
                logger.info("{} Cursor Stopping.", collName);
                cursor.close();
                return;
            }
            ChangeStreamDocument<Document> document = cursor.next();
            if (document.getDocumentKey() == null) {
                logger.info("{} 日志的主键不存在.", collName);
                continue;
            }

            duration = System.currentTimeMillis();
            adminTemplate.saveResumeToken(collName, document.getResumeToken());

            OperationType op = document.getOperationType();
            Map<String, Object> dataMap = new HashMap<>();
            dataMap.put(MgSyncConst.MG_KEY_COLL, collName);
            dataMap.put(MgSyncConst.MG_KEY_OP, op.getValue());

            Map<String, Object> entityProperties = new HashMap<>();
            try {
                if (op == OperationType.DELETE) {
                    BsonValue bsonValue = document.getDocumentKey().get(MgSyncConst.MG_KEY_ID);
                    BsonType bsonType = bsonValue.getBsonType();
                    Object value = null;
                    if (bsonType == BsonType.INT32) {
                        value = bsonValue.asInt32().getValue();
                    } else if (bsonType == BsonType.INT64) {
                        value = bsonValue.asInt64().getValue();
                    } else if (bsonType == BsonType.OBJECT_ID) {
                        value = bsonValue.asObjectId().getValue().toHexString();
                    } else if (bsonType == BsonType.STRING) {
                        value = bsonValue.asString().getValue();
                    }
                    entityProperties.put(MgSyncConst.MG_KEY_ID_0, value);

                    logger.info("coll={}, op={}, id={}", collName, op.getValue(), value);
                } else {
                    if (document.getFullDocument() == null) {
                        logger.info("{} 日志的内容不存在, op={}.", collName, op.getValue());
                        continue;
                    }
                    Iterator<Map.Entry<String, Object>> iterator = document.getFullDocument().entrySet().iterator();
                    while (iterator.hasNext()) {
                        Map.Entry<String, Object> entry = iterator.next();
                        Object value = entry.getValue();
                        if (value != null) {
                            if (value instanceof ObjectId) {
                                value = ((ObjectId) value).toHexString();
                            } else if (value instanceof List) {
                                List<Document> list = (List<Document>) value;
                                for (Document itemDocument : list) {
                                    Iterator<Map.Entry<String, Object>> itemIterator = itemDocument.entrySet().iterator();
                                    while (itemIterator.hasNext()) {
                                        Map.Entry<String, Object> itemEntry = itemIterator.next();
                                        Object itemValue = itemEntry.getValue();
                                        if (itemValue != null) {
                                            if (itemValue instanceof ObjectId) {
                                                itemValue = ((ObjectId) itemValue).toHexString();
                                                itemDocument.put(itemEntry.getKey(), itemValue);
                                            }
                                        }
                                    }
                                }
                            }

                            entityProperties.put(entry.getKey(), value);
                        }
                    }

                    if (entityProperties.containsKey(MgSyncConst.MG_KEY_ID)) {
                        logger.info("coll={}, op={}, id={}", collName, op.getValue(), entityProperties.get(MgSyncConst.MG_KEY_ID));
                    }
                }

                if (entityProperties.size() > 0) {
                    dataMap.put(MgSyncConst.MG_KEY_DATA, entityProperties);
                }

//                curr_token = document.getResumeToken();
//                logger.info("token={}", document.getResumeToken());
                if (SendState == SyncState.committing) {
                    dataBatchBufferList.add(dataMap);
                } else {
                    dataBatchList.add(dataMap);

                    if (dataBatchList.size() >= batchSize) {
                        send();
                    }
                }
            } catch (BsonInvalidOperationException e) {
                logger.error("解析 {} 日志时出错，已跳过.", collName, e);
//            } catch (MongoSocketReadException e) {
//                logger.info("连接断开，等待十秒，期望驱动重连成功");
            }
        }

//        }catch (Exception e) {
//            logger.error("错误", e);
//        }
    }

    private synchronized void send() {
//        if (SendState == SyncState.committing) {
//            return;
//        }
        SendState = SyncState.committing;
        logger.info("Send to MQ collName={}", collName);
        adminTemplate.send(collName, dataBatchList);
//        cache.put(MgSyncConst.KEY + collName, curr_token.toJson());

        dataBatchList = new ArrayList<>();
        duration = 0;
        SendState = SyncState.idle;//需要加锁吗
        if (dataBatchBufferList.size() > 0) {
            dataBatchList.addAll(dataBatchBufferList);
            dataBatchBufferList = new ArrayList<>();
        }

        if(!monitoring) {
            logger.info("{} 如有则发送余量, {}=", collName, dataBatchList.size());
            if(dataBatchList.size() > 0) {
                send();
            }
        }
    }

    @Override
    public WorkerState state() {
        return this.workerState;
    }

    @Override
    public boolean isRunning() {
        return workerState.getState() == SyncState.running;
    }

    @Override
    public boolean isShutdown() {
        return shutdown;
    }

    @Override
    public String name() {
        return this.collName;
    }

    @Override
    public boolean shutdown() {
        shutdown = true;
        cursorClose();
        workerState.setState(SyncState.pause);

//        Thread thread = Thread.currentThread();
//        if (!thread.isAlive() || !thread.isInterrupted()) {
//            thread.interrupt();
//        }

        return true;
    }

    private void cursorClose() {
        monitoring = false;
//        workerState.setState(SyncState.pause);
        if (cursor != null) {
            cursor.close();
            cursor = null;
        }
    }

    public void batchPostTimeoutTasker() {
//        if (workerState.getState() == SyncState.pause) {
//            return;
//        }

        if (SendState != SyncState.committing) {
            if (dataBatchList.size() > 0 &&
                    ((System.currentTimeMillis() - duration) >= batch_timeout)) {
                send();
            }
        }
    }
}
